package com.rrd.hive.udf.brickhouse.udf.collect;

import java.util.UUID;

import org.apache.hadoop.hive.ql.exec.Description;
import org.apache.hadoop.hive.ql.exec.UDF;

@Description(name = "sessionize",
        value = "_FUNC_(string, timestamp) - Returns a session id for the given id and ts(long). Optional third parameter to specify interval tolerance in milliseconds",
        extended = "SELECT _FUNC_(uid, ts), uid, ts, event_type from foo;")

public class SessionizeUDF extends UDF {
    private String lastUid = null;
    private long lastTS = 0;
    private String lastUUID = null;

    public String evaluate(String uid, long ts, int tolerance) {
        if (uid.equals(this.lastUid)
            && this.timeStampCompare(this.lastTS, ts, tolerance)) {
            this.lastTS = ts;
        } else if (uid.equals(this.lastUid)) {
            this.lastTS = ts;
            this.lastUUID = UUID.randomUUID().toString();

        } else {
            this.lastUid = uid;
            this.lastTS = ts;
            this.lastUUID = UUID.randomUUID().toString();
        }
        return this.lastUUID;
    }

    public String evaluate(String uid, long ts) {
        return this.evaluate(uid, ts, 1800000);
    }

    private Boolean timeStampCompare(long lastTS, long ts, int ms) {
        try {
            long difference = ts - lastTS;
            return Math.abs((int) difference) < ms ? true : false;
        } catch (ArithmeticException e) {
            return false;
        }
    }
}